package com.demo.es.service;

import com.alibaba.fastjson.JSON;
import com.demo.es.entity.OrderHeader;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.SearchScrollHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.elasticsearch.core.query.Query;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.stream.Collectors;

/**
 * ES 查询high level方法
 * @author K
 */
@Slf4j
@Service
public class ElasticsearchHighLevelService {

    @Resource
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    /**
     * 通过明细查找订单，sku是嵌套对象内的字段
     * 订单和明细在同一个doc内，明细是嵌套对象。
     * @param skus sku集合
     * @return 订单集合
     */
    public List<OrderHeader> queryByOrderDetailSkus(List<String> skus) {
        TermsQueryBuilder termsQueryBuilder = QueryBuilders.termsQuery("detail.sku", skus);
        NestedQueryBuilder nestedQueryBuilder = QueryBuilders.nestedQuery("detail", termsQueryBuilder, ScoreMode.None);
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.must(nestedQueryBuilder);
        Query query = new NativeSearchQueryBuilder().withQuery(boolQueryBuilder).build();
        SearchHits<OrderHeader> search = elasticsearchRestTemplate.search(query, OrderHeader.class);
        return search.get().map(SearchHit::getContent).collect(Collectors.toList());
    }

    /**
     * 通过sku和name查找订单
     * @param sku 商品编码
     * @param name 商品名
     * @return 订单集合
     */
    public List<OrderHeader> queryByOrderDetailSkuAndName(String sku, String name) {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        boolQueryBuilder.must(QueryBuilders.termsQuery("detail.sku", sku));
        boolQueryBuilder.must(QueryBuilders.termsQuery("detail.name", name));

        NestedQueryBuilder nestedQueryBuilder = QueryBuilders.nestedQuery("detail", boolQueryBuilder, ScoreMode.None);
        Query query = new NativeSearchQueryBuilder().withQuery(nestedQueryBuilder).build();
        SearchHits<OrderHeader> search = elasticsearchRestTemplate.search(query, OrderHeader.class);
        return search.get().map(SearchHit::getContent).collect(Collectors.toList());
    }

    /**
     * BoolQueryBuilder.must 相当于 and
     * @param orderSn 订单号
     * @param address 地址
     * @return 订单集合
     */
    public List<OrderHeader> queryOrderWithCondition(String orderSn, String address, String startPrice,
                                                     String endPrice) {

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        if (StringUtils.isNotBlank(orderSn)) {
            // 精准查询
            boolQueryBuilder.must(QueryBuilders.termQuery("orderSn", orderSn));
        }

        if (StringUtils.isNotBlank(address)) {
            // 模糊查询
            boolQueryBuilder.must(QueryBuilders.wildcardQuery("address", String.format("*%s*", address)));
        }

        if (StringUtils.isNotBlank(startPrice)) {
            // 范围查询，大于
            boolQueryBuilder.must(QueryBuilders.rangeQuery("price").gte(startPrice));
        }

        if (StringUtils.isNotBlank(endPrice)) {
            // 范围查询，小于
            boolQueryBuilder.must(QueryBuilders.rangeQuery("price").lt(endPrice));
        }

        Query query = new NativeSearchQueryBuilder().withQuery(boolQueryBuilder).build();
        SearchHits<OrderHeader> search = elasticsearchRestTemplate.search(query, OrderHeader.class);
        return search.get().map(SearchHit::getContent).collect(Collectors.toList());
    }

    /**
     * BoolQueryBuilder.should 相当于 or
     * @param orderSn 订单号
     * @param address 地址
     * @return 订单集合
     */
    public List<OrderHeader> queryOrderWithShouldCondition(String orderSn, String address) {

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        if (StringUtils.isNotBlank(orderSn)) {
            // 模糊查询
            boolQueryBuilder.should(QueryBuilders.wildcardQuery("orderSn", String.format("*%s*", orderSn)));
        }

        if (StringUtils.isNotBlank(address)) {
            // 模糊查询
            boolQueryBuilder.should(QueryBuilders.wildcardQuery("address", String.format("*%s*", address)));
        }

        Query query = new NativeSearchQueryBuilder().withQuery(boolQueryBuilder).build();

        SearchHits<OrderHeader> search = elasticsearchRestTemplate.search(query, OrderHeader.class);
        log.info("es 结果：{}", JSON.toJSONString(search));
        return search.getSearchHits().stream().map(SearchHit::getContent).collect(Collectors.toList());
    }

    /**
     * es分页， from + size分页查询只能查es前1w条记录,
     * 因为ES使用index.max_result_window:10000作爲保护措施，这个参数可以动态修改，也可以在配置文件配置
     * @param pageNo 当前页码
     * @param pageSize 每页记录数
     * @return 订单集合
     */
    public List<OrderHeader> queryOrderByPage(int pageNo, int pageSize) {

        PageRequest pageRequest = PageRequest.of(pageNo, pageSize);

        Query query = new NativeSearchQueryBuilder().withPageable(pageRequest).build();
        SearchHits<OrderHeader> searchHits = elasticsearchRestTemplate.search(query, OrderHeader.class);
        List<OrderHeader> content = searchHits.get().map(SearchHit::getContent).collect(Collectors.toList());
        log.info("queryOrderByPage es 结果：{}", JSON.toJSONString(content));
        return content;
    }

    /**
     * es分页，使用scroll分页，数据量大，性能会变差，但是不受from + size分页限制。
     * 因为scroll是通过迭代查询数据
     * @param pageNo 当前页码
     * @param pageSize 每页记录数
     * @return 订单集合
     */
    public List<OrderHeader> queryOrderByScroll(int pageNo, int pageSize, String address) {
        // SpringDate ElasticSearch 4.x版本之后 pageNo不能大于0，否则报错
        PageRequest pageRequest = PageRequest.of(0, pageSize);
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        if (StringUtils.isNotBlank(address)) {
            // 精准查询
            boolQueryBuilder.must(QueryBuilders.termQuery("address", address));
        }

        Query query = new NativeSearchQueryBuilder().withQuery(boolQueryBuilder).withPageable(pageRequest).build();

        SearchScrollHits<OrderHeader> searchHits = null;
        List<OrderHeader> content = List.of();
        try {
            // 深度查询分页，Scroll会在es服务器上生成一个快照，保存一定的时间
            searchHits = this.elasticsearchRestTemplate.searchScrollStart(5000, query,
                    OrderHeader.class, IndexCoordinates.of("luna_order"));
            // 通过迭代方式查找指定分页数据
            for (int i = 0; i < pageNo; i++) {
                List<OrderHeader> tmp = searchHits.get().map(SearchHit::getContent).collect(Collectors.toList());
                if (CollectionUtils.isNotEmpty(tmp)) {
                    content.addAll(tmp);
                }
                //取下一页，scrollId在es服务器上可能会发生变化，需要用最新的。发起continueScroll请求会重新刷新快照保留时间
                searchHits = elasticsearchRestTemplate.searchScrollContinue(searchHits.getScrollId(), 30000, OrderHeader.class,
                        IndexCoordinates.of("luna-order"));
            }
        } finally {
            if (searchHits != null) {
                // 最后释放查询
                elasticsearchRestTemplate.searchScrollClear(List.of(searchHits.getScrollId()));
            }
        }
        log.info("queryOrderByScroll es 结果：{}", JSON.toJSONString(content));

        return content;
    }

    /**
     * Search After 分页方式是 ES 5 新增的一种分页查询方式，其实现的思路同 Scroll 分页方式基本一致，通过记录上一次分页的位置标识，
     * 来进行下一次分页数据的查询。相比于 Scroll 分页方式，它的优点是可以实时体现数据的变化，解决了查询快照导致的查询结果延迟问题。
     * @param pageNo 当前页码
     * @param pageSize 每页记录数
     * @param address
     * @return 订单集合
     */
    public List<OrderHeader> queryOrderBySearchAfter(int pageNo, int pageSize, String address) {

        // SpringDate ElasticSearch 4.x版本之后 pageNo不能大于0，否则报错
        PageRequest pageRequest = PageRequest.of(0, pageSize);
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        if (StringUtils.isNotBlank(address)) {
            // 精准查询
            boolQueryBuilder.must(QueryBuilders.termQuery("address", address));
        }

        FieldSortBuilder fieldSortBuilder = new FieldSortBuilder("orderId");
        fieldSortBuilder.order(SortOrder.ASC);

        Query query = new NativeSearchQueryBuilder().withQuery(boolQueryBuilder).withPageable(pageRequest)
                .withSearchType(SearchType.DFS_QUERY_THEN_FETCH)
                // 必须进行排序才能在搜索结果拿到sortValues的值
                .withSorts(fieldSortBuilder)
                .build();

        List<OrderHeader> content = List.of();
        SearchHits<OrderHeader> searchHits = this.elasticsearchRestTemplate.search(query, OrderHeader.class);
        // 通过迭代方式查找指定分页数据
        for (int i = 0; i < pageNo; i++) {
            if (searchHits.getSearchHits().isEmpty()) {
                break;
            }

            content = searchHits.get().map(SearchHit::getContent).collect(Collectors.toList());

            List<Object> sortValues = searchHits.getSearchHit(searchHits.getSearchHits().size() - 1).getSortValues();
            log.info("sortValues:{}", sortValues);
            query.setSearchAfter(sortValues);

            //取下一页，通过sortValues获取最新数据
            searchHits = elasticsearchRestTemplate.search(query, OrderHeader.class);
        }

        log.info("queryOrderByScroll es 结果：{}", JSON.toJSONString(content));

        return content;
    }

    /**
     * 查询结果以关键字开头(优先级最高)进行排序
     * es使用ik分词器将中文词组，因为es默认分词器将中文搜索词拆成一个个汉字
     * @param keyword 查询条件
     * @return 查询结果
     */
    public List<OrderHeader> disMaxQuery(String keyword) {
        DisMaxQueryBuilder disMaxQueryBuilder = QueryBuilders.disMaxQuery();
        MatchQueryBuilder matchQueryBuilder = QueryBuilders
                // 将搜索词分词，再与目标查询字段进行匹配，若分词中的任意一个词与目标字段匹配上，则可查询到
                .matchQuery("address", keyword);
        disMaxQueryBuilder.add(matchQueryBuilder);
        Query query = new NativeSearchQueryBuilder().withQuery(disMaxQueryBuilder).build();
        SearchHits<OrderHeader> searchHits = elasticsearchRestTemplate.search(query, OrderHeader.class);
        return searchHits.get().map(SearchHit::getContent).collect(Collectors.toList());
    }
}
